// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: jshin

#pragma once
#include <functional>
#include <vector>
#include <map>
#include <unordered_map>
#include <memory>
#include <set>
#include <string>
#include <string_view>
#include <utility>

#include "coin2/exchange/base/executor/topic_publisher.h"
#include "coin2/exchange/base/feed/distributor.h"
#include "coin2/exchange/base/feed/update.h"
#include "coin2/exchange/base/feed_util/book_builder.h"
#include "coin2/exchange/base/feed_util/trade.h"
#include "coin2/exchange/base/feed_util/wallet.h"
#include "coin2/exchange/base/feed_util/market_analysis.h"

namespace coin2::exchange::base::feed {
namespace impl {

using coin2::exchange::base::executor::TopicRecordPacket;
using coin2::exchange::base::symbology::IProduct;

class FeedUpdateManager {
 public:
  explicit FeedUpdateManager(
      const std::string& system_name,
      const MarketExchangeApi& mea,
      bool enable_diff_book,
      bool enable_telemetry)
      : packet_subscriber_(nullptr),
        upd_(system_name, mea),
        enable_diff_book_(enable_diff_book),
        enable_telemetry_(enable_telemetry) {
    if (enable_telemetry) {
      feed_distributor_.reset(new FeedDistributor(&(upd_.MutableTs())));
    } else {
      feed_distributor_.reset(new FeedDistributor(nullptr));
    }
  }

  void SetRecipe(const std::string& recipe) { recipe_ = recipe; }
  const FeedUpdate& update() const { return upd_; }
  FeedUpdate* mutable_update() { return &upd_; }
  const std::string& system_name() const { return upd_.system_name(); }
  const MarketExchangeApi& mea() const { return upd_.mea(); }
  void SetDistibutor(std::unique_ptr<FeedDistributor> distributor) {
    feed_distributor_.reset();
    feed_distributor_ = std::move(distributor);
  }

  const std::set<std::string, std::less<>>& native_symbols() const {
    return feed_distributor_->native_symbols();
  }

  bool is_symbol_registered(std::string_view native_symbol) const {
    return feed_distributor_->is_symbol_registered(native_symbol);
  }

  const std::vector<std::unique_ptr<IProduct>>& products() const {
    return products_;
  }

  void AddProductForFeed(const IProduct& product, IFeedSubscriber* subscriber) {
    // not product.native_symbol();
    const auto& native_symbol = product.feed_native_symbol();
    feed_distributor_->AddProduct(product, subscriber);
    books_[native_symbol] = std::make_unique<BookBuilder>(enable_diff_book_);
    slim_trades_[native_symbol] = SlimTrade();
    products_.emplace_back(product.Clone());
    product_map_[native_symbol].reset(product.Clone());
    book_exchange_ts_map_[native_symbol] = 0;
    trade_exchange_ts_map_[native_symbol] = 0;
    liquidation_order_exchange_ts_map_[native_symbol] = 0;
  }

  void RemoveProductForFeed(const std::vector<const IProduct*>& products) {
    // only remove symbol from distributor, other data may be referenced
    feed_distributor_->RemoveProduct(products);
  }

  void SetPrePostReadStats(const TopicRecordPacket& packet) {
    auto& ts = upd_.MutableTs();
    ts.set_pre_read_available(packet.pre_read_available);
    ts.set_pre_read_timestamp(packet.pre_read_timestamp);
    ts.set_post_read_available(packet.post_read_available);
    ts.set_post_read_timestamp(packet.timestamp);
  }

 public:  // distributor
  void onBeginPacket(const TopicRecordPacket& packet) {
    upd_.Reset();
    upd_.SetRecipe(recipe_);
    upd_.SetTimestamp(packet.timestamp);
    if (enable_telemetry_) {
      auto& ts = upd_.MutableTs();
      ts.set_raw_rx_timestamp(packet.timestamp);  // FIX
      ts.set_raw_tx_timestamp(packet.timestamp);
      ts.set_main_rx_timestamp(packet.distributor_rx_timestamp);
    }
    if (packet_subscriber_) {
      packet_subscriber_->onBeginPacket(upd_);
    }
  }

  void onEndPacket(const TopicRecordPacket& packet) {
    if (packet_subscriber_) {
      packet_subscriber_->onEndPacket(upd_);
    }
  }

  BookBuilder* GetBookMutable(std::string_view native_symbol) {
    auto iter = product_map_.find(native_symbol);
    if (iter == product_map_.end()) {
      return nullptr;
    }
    upd_.SetType(native_symbol, *(iter->second.get()), coin2::exchange::feed::BOOK_FEED);
    auto& book = books_.find(native_symbol)->second;
    upd_.SetBook(book.get());
    return book.get();
  }

  const BookBuilder* GetBook(std::string_view native_symbol) const {
    auto iter = product_map_.find(native_symbol);
    if (iter == product_map_.end()) {
      return nullptr;
    }
    auto& book = books_.find(native_symbol)->second;
    return book.get();
  }

  BookBuilder* GetBookMutable(std::string_view native_symbol, const std::string& type) {
    auto iter = product_map_.find(native_symbol);
    if (iter == product_map_.end()) {
      return nullptr;
    }
    upd_.SetType(native_symbol, *(iter->second.get()), coin2::exchange::feed::BOOK_FEED);
    // TODO(jshin): add type based map
    auto& book = books_.find(native_symbol)->second;
    upd_.SetBook(book.get());
    return book.get();
  }

  Trade* GetTradeMutable() { return &trade_; }

  SlimTrade* GetSlimTradeMutable(std::string_view native_symbol) {
    auto iter = product_map_.find(native_symbol);
    if (iter == product_map_.end()) {
      return nullptr;
    }
    auto& slim_trade = slim_trades_.find(native_symbol)->second;
    return &slim_trade;
  }

  LiquidationOrder* GetLiquidationOrderMutable() { return &liquidation_order_; }

  OpenInterest* GetOpenInterestMutable() { return &open_interest_; }

  MarkPrice* GetMarkPriceMutable() { return &mark_price_; }

  Index* GetIndexMutable() { return &index_; }

  FundingRate* GetFundingRateMutable() { return &funding_rate_; }

  IvGreeks* GetIvGreeksMutable() { return &iv_greeks_; }

  Nav* GetNavMutable() { return &nav_; }

  Wallet* GetWalletMutable() { return &wallet_; }

  MarketAnalysis* GetMarketAnalysisMutable() { return &market_analysis_; }

  TradingData<TopLongShortAccountRatio>* GetTopLongShortAccountRatioMutable(
      std::string_view native_symbol,
      fastfeed::proto::TradingDataInterval interval) {
    auto iter = product_map_.find(native_symbol);
    if (iter == product_map_.end()) {
      return nullptr;
    }
    auto* ratios = &(top_long_short_account_ratio_map_[std::make_pair(std::string(native_symbol), interval)]);
    ratios->SetInterval(interval);
    upd_.SetType(native_symbol, *(iter->second.get()), coin2::exchange::feed::STATUS_FEED);
    upd_.SetTopLongShortAccountRatio(ratios);
    return ratios;
  }

  TradingData<TopLongShortPositionRatio>* GetTopLongShortPositionRatioMutable(
      std::string_view native_symbol,
      fastfeed::proto::TradingDataInterval interval) {
    auto iter = product_map_.find(native_symbol);
    if (iter == product_map_.end()) {
      return nullptr;
    }
    auto* ratios = &(top_long_short_position_ratio_map_[std::make_pair(std::string(native_symbol), interval)]);
    ratios->SetInterval(interval);
    upd_.SetType(native_symbol, *(iter->second.get()), coin2::exchange::feed::STATUS_FEED);
    upd_.SetTopLongShortPositionRatio(ratios);
    return ratios;
  }

  TradingData<GlobalLongShortAccountRatio>* GetGlobalLongShortAccountRatioMutable(
      std::string_view native_symbol,
      fastfeed::proto::TradingDataInterval interval) {
    auto iter = product_map_.find(native_symbol);
    if (iter == product_map_.end()) {
      return nullptr;
    }
    auto* ratios = &(global_long_short_account_ratio_map_[std::make_pair(std::string(native_symbol), interval)]);
    ratios->SetInterval(interval);
    upd_.SetType(native_symbol, *(iter->second.get()), coin2::exchange::feed::STATUS_FEED);
    upd_.SetGlobalLongShortAccountRatio(ratios);
    return ratios;
  }

  TradingData<TakerBuySellRatio>* GetTakerBuySellRatioMutable(
      std::string_view native_symbol,
      fastfeed::proto::TradingDataInterval interval) {
    auto iter = product_map_.find(native_symbol);
    if (iter == product_map_.end()) {
      return nullptr;
    }
    auto* ratios = &(taker_long_short_ratio_map_[std::make_pair(std::string(native_symbol), interval)]);
    ratios->SetInterval(interval);
    upd_.SetType(native_symbol, *(iter->second.get()), coin2::exchange::feed::STATUS_FEED);
    upd_.SetTakerBuySellRatio(ratios);
    return ratios;
  }

  TradingData<OpenInterest>* GetOpenInterestHistMutable(
      std::string_view native_symbol,
      fastfeed::proto::TradingDataInterval interval) {
    auto iter = product_map_.find(native_symbol);
    if (iter == product_map_.end()) {
      return nullptr;
    }
    auto* open_interest_hist = &(open_interest_hist_map_[std::make_pair(std::string(native_symbol), interval)]);
    open_interest_hist->SetInterval(interval);
    upd_.SetType(native_symbol, *(iter->second.get()), coin2::exchange::feed::STATUS_FEED);
    upd_.SetOpenInterestHist(open_interest_hist);
    return open_interest_hist;
  }

  Klines* GetKlineMutable(std::string_view native_symbol, fastfeed::proto::KlineInterval interval) {
    auto iter = product_map_.find(native_symbol);
    if (iter == product_map_.end()) {
      return nullptr;
    }
    Klines* klines = &(klines_[std::make_pair(std::string(native_symbol), interval)]);

    upd_.SetType(native_symbol, *(iter->second.get()), coin2::exchange::feed::STATUS_FEED);
    upd_.SetKlines(klines);
    return klines;
  }

  void PublishLiquidationOrder(std::string_view native_symbol, bool has_more_liquidation_order) {
    liquidation_order_.has_more_liquidation_order = has_more_liquidation_order;
    upd_.SetLiquidationOrder(&liquidation_order_);
    auto& product = product_map_.find(native_symbol)->second;
    upd_.SetType(native_symbol, *(product.get()), coin2::exchange::feed::STATUS_FEED);
    upd_.MutableTs().set_exchange_publish_timestamp(upd_.liquidation_order().timestamp);
    feed_distributor_->onLiquidationFeed(upd_.native_symbol(), upd_);
  }

  void PublishLiquidationOrderWithExchangeTimeCheck(
      const std::string& native_symbol,
      bool has_more_liquidation_order) {
    if (liquidation_order_.timestamp > 0 &&
        liquidation_order_.timestamp < liquidation_order_exchange_ts_map_[native_symbol]) {
      LOG(WARNING) << "Liquidation order of " << native_symbol << " ET timestamp inverse! "
                   << liquidation_order_.timestamp << ","
                   << liquidation_order_exchange_ts_map_[native_symbol] << ","
                   << upd_.timestamp();
      return;
    } else {
      liquidation_order_exchange_ts_map_[native_symbol] = liquidation_order_.timestamp;
    }
    PublishLiquidationOrder(native_symbol, has_more_liquidation_order);
  }

  void PublishOpenInterest(std::string_view native_symbol) {
    upd_.SetOpenInterest(&open_interest_);
    auto& product = product_map_.find(native_symbol)->second;
    upd_.SetType(native_symbol, *(product.get()), coin2::exchange::feed::STATUS_FEED);
    upd_.MutableTs().set_exchange_publish_timestamp(upd_.open_interest().timestamp);
    feed_distributor_->onOpenInterestFeed(upd_.native_symbol(), upd_);
  }

  void PublishOpenInterestWithExchangeTimeCheck(const std::string& native_symbol) {
    if (open_interest_.timestamp > 0 &&
        open_interest_.timestamp <= open_interest_exchange_ts_map_[native_symbol]) {
      static int occurs = 0;
      occurs++;
      if (occurs <= 100) {
        LOG(WARNING) << "Open interest of " << native_symbol << " ET timestamp inverse! "
                    << open_interest_.timestamp << ","
                    << open_interest_exchange_ts_map_[native_symbol] << ","
                    << upd_.timestamp();
      }
      return;
    } else {
      open_interest_exchange_ts_map_[native_symbol] = open_interest_.timestamp;
    }
    PublishOpenInterest(native_symbol);
  }

  void PublishMarkPrice(std::string_view native_symbol) {
    upd_.SetMarkPrice(&mark_price_);
    auto& product = product_map_.find(native_symbol)->second;
    upd_.SetType(native_symbol, *(product.get()), coin2::exchange::feed::STATUS_FEED);
    upd_.MutableTs().set_exchange_publish_timestamp(upd_.mark_price().timestamp);
    feed_distributor_->onMarkPriceFeed(upd_.native_symbol(), upd_);
  }

  void PublishMarkPriceWithExchangeTimeCheck(const std::string& native_symbol) {
    if (mark_price_.timestamp > 0 &&
        mark_price_.timestamp < mark_price_exchange_ts_map_[native_symbol]) {
      LOG(WARNING) << "Mark price of " << native_symbol << " ET timestamp inverse! "
                   << mark_price_.timestamp << ","
                   << mark_price_exchange_ts_map_[native_symbol] << ","
                   << upd_.timestamp();
      return;
    } else {
      mark_price_exchange_ts_map_[native_symbol] = mark_price_.timestamp;
    }
    PublishMarkPrice(native_symbol);
  }

  void PublishIndex(std::string_view native_symbol) {
    upd_.SetIndex(&index_);
    auto& product = product_map_.find(native_symbol)->second;
    upd_.SetType(native_symbol, *(product.get()), coin2::exchange::feed::STATUS_FEED);
    upd_.MutableTs().set_exchange_publish_timestamp(upd_.index().timestamp);
    feed_distributor_->onIndexFeed(upd_.native_symbol(), upd_);
  }

  void PublishIndexWithExchangeTimeCheck(const std::string& native_symbol) {
    if (index_.timestamp > 0 &&
        index_.timestamp <= index_exchange_ts_map_[native_symbol]) {
      static int occurs = 0;
      occurs++;
      if (occurs <= 100) {
        LOG(WARNING) << "Index of " << native_symbol << " ET timestamp inverse! "
                    << index_.timestamp << ","
                    << index_exchange_ts_map_[native_symbol] << ","
                    << upd_.timestamp();
      }
      return;
    } else {
      index_exchange_ts_map_[native_symbol] = index_.timestamp;
    }
    PublishIndex(native_symbol);
  }

  void PublishFundingRate(std::string_view native_symbol) {
    upd_.SetFundingRate(&funding_rate_);
    auto& product = product_map_.find(native_symbol)->second;
    upd_.SetType(native_symbol, *(product.get()), coin2::exchange::feed::STATUS_FEED);
    upd_.MutableTs().set_exchange_publish_timestamp(upd_.funding_rate().timestamp);
    feed_distributor_->onFundingRateFeed(upd_.native_symbol(), upd_);
  }

  void PublishFundingRateWithExchangeTimeCheck(const std::string& native_symbol) {
    if (funding_rate_.timestamp > 0 &&
        funding_rate_.timestamp < funding_rate_exchange_ts_map_[native_symbol]) {
      LOG(WARNING) << "Funding rate of " << native_symbol << " ET timestamp inverse! "
                   << funding_rate_.timestamp << ","
                   << funding_rate_exchange_ts_map_[native_symbol] << ","
                   << upd_.timestamp();
      return;
    } else {
      funding_rate_exchange_ts_map_[native_symbol] = funding_rate_.timestamp;
    }
    PublishFundingRate(native_symbol);
  }

  void PublishIvGreeks(std::string_view native_symbol) {
    upd_.SetIvGreeks(&iv_greeks_);
    auto& product = product_map_.find(native_symbol)->second;
    upd_.SetType(native_symbol, *(product.get()), coin2::exchange::feed::STATUS_FEED);
    upd_.MutableTs().set_exchange_publish_timestamp(upd_.iv_greeks().timestamp);
    feed_distributor_->onIvGreeksFeed(upd_.native_symbol(), upd_);
  }

  void PublishIvGreeksWithExchangeTimeCheck(const std::string& native_symbol) {
    if (iv_greeks_.timestamp > 0 &&
        iv_greeks_.timestamp < iv_greeks_exchange_ts_map_[native_symbol]) {
      LOG(WARNING) << "Iv Greeks of " << native_symbol << " ET timestamp inverse! "
                   << iv_greeks_.timestamp << ","
                   << iv_greeks_exchange_ts_map_[native_symbol] << ","
                   << upd_.timestamp();
      return;
    } else {
      iv_greeks_exchange_ts_map_[native_symbol] = iv_greeks_.timestamp;
    }
    PublishIvGreeks(native_symbol);
  }

  void PublishNav(std::string_view native_symbol) {
    upd_.SetNav(&nav_);
    auto& product = product_map_.find(native_symbol)->second;
    upd_.SetType(native_symbol, *(product.get()), coin2::exchange::feed::STATUS_FEED);
    upd_.MutableTs().set_exchange_publish_timestamp(upd_.nav().timestamp);
    feed_distributor_->onNavFeed(upd_.native_symbol(), upd_);
  }

  void PublishNavWithExchangeTimeCheck(const std::string& native_symbol) {
    if (nav_.timestamp > 0 &&
        nav_.timestamp < nav_exchange_ts_map_[native_symbol]) {
      LOG(WARNING) << "Nav of " << native_symbol << " ET timestamp inverse! "
                   << nav_.timestamp << ","
                   << nav_exchange_ts_map_[native_symbol] << ","
                   << upd_.timestamp();
      return;
    } else {
      nav_exchange_ts_map_[native_symbol] = nav_.timestamp;
    }
    PublishNav(native_symbol);
  }

  void PublishKline(std::string_view native_symbol) {
    upd_.MutableTs().set_exchange_publish_timestamp(upd_.klines().Get().value()->timestamp);
    feed_distributor_->onKlineFeed(upd_.native_symbol(), upd_);
  }

  void PublishTrade(std::string_view native_symbol, bool has_more_trade) {
    trade_.has_more_trade = has_more_trade;
    upd_.SetTrade(&trade_);
    auto& product = product_map_.find(native_symbol)->second;
    upd_.SetType(native_symbol, *(product.get()), coin2::exchange::feed::TRADE_FEED);
    upd_.MutableTs().set_exchange_publish_timestamp(upd_.trade().timestamp);
    CHECK_NE(trade_.side, 0);
    feed_distributor_->onTradeFeed(upd_.native_symbol(), upd_);
  }

  void PublishTradeWithExchangeTimeCheck(
      const std::string& native_symbol,
      bool has_more_trade) {
    if (trade_.timestamp > 0 &&
        trade_.timestamp < trade_exchange_ts_map_[native_symbol]) {
      SPDLOG_WARN_EVERY_NS(
          5'000'000'000L,
          upd_.timestamp(),
          "{} {} {} {}, {}, {}",
          "Trade of",
          native_symbol,
          "ET timestamp inverse!",
          trade_.timestamp,
          trade_exchange_ts_map_[native_symbol],
          upd_.timestamp());
      return;
    } else {
      trade_exchange_ts_map_[native_symbol] = trade_.timestamp;
    }
    PublishTrade(native_symbol, has_more_trade);
  }

  void PublishBook() {
    upd_.MutableTs().set_exchange_publish_timestamp(upd_.book().Timestamp());
    upd_.SetType(coin2::exchange::feed::BOOK_FEED);
    feed_distributor_->onBookFeed(upd_.native_symbol(), upd_);
  }

  bool BookExchangeTimeCheck() {
    std::string native_symbol = std::string(upd_.native_symbol());
    if (upd_.book().Timestamp() > 0 &&
        upd_.book().Timestamp() < book_exchange_ts_map_[native_symbol]) {
      SPDLOG_WARN_EVERY_NS(
          5'000'000'000L,
          upd_.timestamp(),
          "{} {} {} {}, {}, {}",
          "Book of",
          native_symbol,
          "ET timestamp inverse!",
          upd_.book().Timestamp(),
          book_exchange_ts_map_[native_symbol],
          upd_.timestamp());
      return false;
    } else {
      book_exchange_ts_map_[native_symbol] = upd_.book().Timestamp();
      return true;
    }
  }

  void PublishBookWithExchangeTimeCheck() {
    if (!BookExchangeTimeCheck()) {
      return;
    }
    PublishBook();
  }

  void PublishWallet(std::string_view native_symbol) {
    upd_.SetWallet(&wallet_);
    auto& product = product_map_.find(native_symbol)->second;
    upd_.SetType(native_symbol, *(product.get()), coin2::exchange::feed::STATUS_FEED);
    upd_.MutableTs().set_exchange_publish_timestamp(upd_.wallet().timestamp);
    feed_distributor_->onWalletFeed(upd_.native_symbol(), upd_);
  }

  const std::string& GetRelativeNorm(const std::string& native_symbol) {
    return product_map_[native_symbol]->relative_norm();
  }

  const std::string& GetQuoteCurrency(const std::string& native_symbol) {
    return product_map_[native_symbol]->quote().currency();
  }

  void PublishMarketAnalysis(std::string_view native_symbol) {
    upd_.SetMarketAnalysis(&market_analysis_);
    auto& product = product_map_.find(native_symbol)->second;
    upd_.SetType(native_symbol, *(product.get()), coin2::exchange::feed::STATUS_FEED);
    upd_.MutableTs().set_exchange_publish_timestamp(upd_.market_analysis().timestamp);
    feed_distributor_->onMarketAnalysisFeed(upd_.native_symbol(), upd_);
  }

  void PublishTopLongShortAccountRatio() {
    upd_.MutableTs().set_exchange_publish_timestamp(upd_.top_long_short_account_ratio().Timestamp());
    feed_distributor_->onTopLongShortAccountRatioFeed(upd_.native_symbol(), upd_);
  }

  void PublishTopLongShortPositionRatio() {
    upd_.MutableTs().set_exchange_publish_timestamp(upd_.top_long_short_position_ratio().Timestamp());
    feed_distributor_->onTopLongShortPositionRatioFeed(upd_.native_symbol(), upd_);
  }

  void PublishGlobalLongShortAccountRatio() {
    upd_.MutableTs().set_exchange_publish_timestamp(upd_.global_long_short_account_ratio().Timestamp());
    feed_distributor_->onGlobalLongShortAccountRatioFeed(upd_.native_symbol(), upd_);
  }

  void PublishTakerBuySellRatio() {
    upd_.MutableTs().set_exchange_publish_timestamp(upd_.taker_buy_sell_ratio().Timestamp());
    feed_distributor_->onTakerBuySellRatioFeed(upd_.native_symbol(), upd_);
  }

  void PublishOpenInterestHist() {
    upd_.MutableTs().set_exchange_publish_timestamp(upd_.open_interest_hist().Timestamp());
    feed_distributor_->onOpenInterestHistFeed(upd_.native_symbol(), upd_);
  }

 protected:
  IPacketSubscriber* packet_subscriber_;
  std::unique_ptr<FeedDistributor> feed_distributor_;

  FeedUpdate upd_;
  bool enable_diff_book_;
  bool enable_telemetry_;
  std::map<std::string, std::unique_ptr<BookBuilder>, std::less<>> books_;
  Trade trade_{};
  std::map<std::string, SlimTrade, std::less<>> slim_trades_;
  LiquidationOrder liquidation_order_{};
  OpenInterest open_interest_{};
  MarkPrice mark_price_{};
  Index index_{};
  FundingRate funding_rate_{};
  IvGreeks iv_greeks_{};
  Nav nav_{};
  Wallet wallet_{};
  MarketAnalysis market_analysis_{};
  std::map<std::pair<std::string, fastfeed::proto::TradingDataInterval>,
      TradingData<TopLongShortAccountRatio>> top_long_short_account_ratio_map_{};
  std::map<std::pair<std::string, fastfeed::proto::TradingDataInterval>,
      TradingData<TopLongShortPositionRatio>> top_long_short_position_ratio_map_{};
  std::map<std::pair<std::string, fastfeed::proto::TradingDataInterval>,
      TradingData<GlobalLongShortAccountRatio>> global_long_short_account_ratio_map_{};
  std::map<std::pair<std::string, fastfeed::proto::TradingDataInterval>,
      TradingData<TakerBuySellRatio>> taker_long_short_ratio_map_{};
  std::map<std::pair<std::string, fastfeed::proto::TradingDataInterval>,
      TradingData<OpenInterest>> open_interest_hist_map_{};
  // <symbol, interval> => klines
  std::map<std::pair<std::string, fastfeed::proto::KlineInterval>, Klines> klines_;
  // FIX: avoid unique_ptr and share instances.
  std::vector<std::unique_ptr<IProduct>> products_;
  std::map<std::string, std::unique_ptr<IProduct>, std::less<>> product_map_;

  std::unordered_map<std::string, int64_t> book_exchange_ts_map_;
  std::unordered_map<std::string, int64_t> trade_exchange_ts_map_;
  std::unordered_map<std::string, int64_t> liquidation_order_exchange_ts_map_;
  std::unordered_map<std::string, int64_t> open_interest_exchange_ts_map_;
  std::unordered_map<std::string, int64_t> mark_price_exchange_ts_map_;
  std::unordered_map<std::string, int64_t> index_exchange_ts_map_;
  std::unordered_map<std::string, int64_t> funding_rate_exchange_ts_map_;
  std::unordered_map<std::string, int64_t> nav_exchange_ts_map_;
  std::unordered_map<std::string, int64_t> iv_greeks_exchange_ts_map_;
  std::string recipe_;
};

}  // namespace impl

using impl::FeedUpdateManager;
}  // namespace coin2::exchange::base::feed
